package chaoyue.study.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class MyKafkaListener {
    /**
     * topics：指定要监听的Topic
     * groupId：指定消费者GroupId
     * 默认的监听是从最后一个offset开始的
     */
    @KafkaListener(topics = "news", groupId = "testGroup")
    public void doSome(ConsumerRecord record) {
        // 通过ConsumerRecord可以获取详细的详细信息
        Object key = record.key();
        Object value = record.value();
        log.info("收到消息-1 key:{}, value:{}", key, value);
    }

    /**
     * 消费者还可以指定手动指定分区和偏移量
     */
    @KafkaListener(groupId = "testGroup1", topicPartitions = {
                    @TopicPartition(topic = "news", partitionOffsets =
                    @PartitionOffset(partition = "0", initialOffset = "0"))
            })
    public void doSome2(ConsumerRecord record) {
        // 通过ConsumerRecord可以获取详细的详细信息
        Object key = record.key();
        Object value = record.value();
        log.info("收到消息-2 key:{}, value:{}", key, value);
    }
}
